#include "kafka_consumer.h"
#include "kafka_producer.h"

#include <cstdarg>
#include <iomanip>
#include <iostream>
#include <memory>
using namespace std;

#define GLOG_USE_GLOG_EXPORT
#include <glog/logging.h>

#define PRODUCER_THREAD_NUM 2
#define KAFKA_TOPIC_NAME "RealTimeTopic"

std::atomic<uint64_t> g_messageId(0);

int sprintf_safe(char *dest, int size, const char *fmt, ...)
{
    va_list args;
    va_start(args, fmt);
#if defined(__GNUC__)
    int n = vsprintf(dest, fmt, args);
#elif defined(_MSC_VER)
    int n = vsprintf_s(dest, size, fmt, args);
#endif
    va_end(args);
    return n;
}

std::string get_current_format_datetime()
{
    auto   tp    = std::chrono::system_clock::now();
    time_t timer = std::chrono::system_clock::to_time_t(tp);

    std::ostringstream oss;
    char               datetime[32] = {0};
    std::strftime(datetime, 32, "%Y-%m-%d %H:%M:%S", std::localtime(&timer));

    auto micros   = std::chrono::duration_cast<std::chrono::microseconds>(tp.time_since_epoch());
    auto second   = std::chrono::duration_cast<std::chrono::seconds>(tp.time_since_epoch());
    auto microsec = (micros - second).count();
    oss << datetime << "." << std::setfill('0') << std::setw(6) << microsec;
    return oss.str();
}

void test_pub_thread(std::shared_ptr<KafkaProducer> producer)
{
    thread::id threadId = this_thread::get_id();

    while (true) {
        char        message[1024] = {0};
        std::string currentTime   = get_current_format_datetime();
        sprintf(message, "[%s]: Hello, %ld!", currentTime.c_str(), g_messageId.fetch_add(1));

        if (!producer->produce(KAFKA_TOPIC_NAME, message)) {
            LOG(WARNING) << "线程ID: " << threadId << ", 消息推送失败";
        } else {
            LOG(INFO) << "线程ID: " << threadId << ", 消息推送成功: " << message;
        }
        this_thread::sleep_for(chrono::milliseconds(1000));
    }
}

void test_sub_thread(string groupId)
{
    LOG(INFO) << "consumer start: " << groupId;
    auto consumer = std::make_shared<KafkaConsumer>("127.0.0.1:9092", groupId);
    consumer->init(true, "sasl_plaintext", "SCRAM-SHA-256", "alice", "alice-secret");
    consumer->subscribe(KAFKA_TOPIC_NAME, 0, "end");
    for (;;) {
        std::shared_ptr<RdKafka::Message> msg = consumer->consume(100);
        if (msg) {
            LOG(INFO) << groupId << " recv: " << static_cast<const char *>(msg->payload());
        }
    }
    LOG(INFO) << "consumer exit: " << groupId;
    consumer->unsubscribe();
}

void consumer()
{
    std::vector<std::string> groupIds = {"test1"};
    thread                   threads[groupIds.size()];

    for (int i = 0; i < groupIds.size(); i++) {
        threads[i] = thread(&test_sub_thread, groupIds[i]);
    }
    for (int i = 0; i < groupIds.size(); i++) {
        threads[i].join();
    }
}

void producer()
{
    auto producer = std::make_shared<KafkaProducer>("127.0.0.1:9092");
    producer->init(true, "sasl_plaintext", "SCRAM-SHA-256", "alice", "alice-secret");

    std::thread threads[PRODUCER_THREAD_NUM];
    for (int i = 0; i < PRODUCER_THREAD_NUM; i++) {
        threads[i] = thread(&test_pub_thread, producer);
    }
    for (int i = 0; i < PRODUCER_THREAD_NUM; i++) {
        threads[i].join();
    }
}

int main(int argc, char *argv[])
{
    if (argc < 2) {
        LOG(WARNING) << "Usage: " << argv[0] << " consumer/producer";
        return 0;
    }

    if (strcmp(argv[1], "consumer") == 0) {
        consumer();
    } else if (strcmp(argv[1], "producer") == 0) {
        producer();
    } else {
        LOG(WARNING) << "Usage: " << argv[0] << " consumer/producer";
    }
    return 0;
}